package com.chief.sink.redis;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import redis.clients.jedis.JedisCluster;

public class RedisSource extends RichParallelSourceFunction<String> {

    JedisCluster jc = null;
    boolean isRuning = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        jc = RedisTest.getJedisCluster();
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRuning) {
            for (int i = 0; i < 1000; i++) {
                ctx.collect(jc.get("index_" + i));
                jc.del("index_" + i);
                Thread.sleep(10000);
            }
        }
    }

    @Override
    public void cancel() {
        isRuning = false;
    }


    @Override
    public void close() throws Exception {
        jc.close();
        jc = null;
    }
}
